package com.bbx.flink.table_api_or_sql;


import com.bbx.flink.demo.entity.Person;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 *读取表中数据写入 数据库
 * 注：有主键时，会对数据进行更新
 */
@Slf4j
public class ConnectorTableJDBCDemo {


    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        String path = "C:\\Users\\kaifacs\\Desktop\\aaaa.txt";
        //csv 格式   用Excec 创建的   第一行数据有问题，使用sql 导出的 csv 测试没有问题
//        String path = "C:\\Users\\kaifacs\\Desktop\\202011040001.csv";
        SingleOutputStreamOperator<Person> dataStream = env.readTextFile(path, "UTF-8")
                .map(i -> {
                    log.info("~~~~~~~~~~~[{}]",i);
                    String[] split = i.split(",");
                    return new Person(split[0], split[1], split[2]);
                });


        tableEnv.createTemporaryView("sourcrTable", dataStream);
        String createSql = "CREATE TABLE MyUserTable (\n" +
                "id String,\n" +
                "name String,  \n" +
                "phone String, \n" +
                "total bigint ,\n" +
                "PRIMARY KEY (id)  NOT ENFORCED \n    " +
                ") WITH (\n" +
                "  'connector.type' = 'jdbc',\n" +
                "  'connector.url' = 'jdbc:mysql://192.168.10.7:3306/flink_user?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC',\n" +
                "  'connector.table' = 'person', \n" +
                "  'connector.driver' = 'com.mysql.cj.jdbc.Driver',\n" +
                "  'connector.username' = 'root',\n" +
                "  'connector.password' = '123456'\n" +
                ")";
        tableEnv.executeSql(createSql);
        String insertSql = " insert   into  MyUserTable  select id,name,phone,count(*) total " +
                "from  sourcrTable  group by id ,name,phone ";
        tableEnv.executeSql(insertSql);

        env.execute();
    }

}
